package com.auto.server;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

import com.auto.storm.SocketBolt;
import com.auto.storm.SocketCountBolt;
import com.auto.storm.SocketSpout;

public class StartTopology {
	
	public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, InterruptedException, AuthorizationException {
			     // configure & build topology
			     TopologyBuilder builder = new TopologyBuilder();
			     String[] records = new String[] {
			               "A Storm cluster is superficially similar to a Hadoop cluster",
			               "All coordination between Nimbus and the Supervisors is done through a Zookeeper cluster",
			               "The core abstraction in Storm is the stream"
			     };
			     builder
			          .setSpout("spout-producer", new SocketSpout(),1)
			          .setNumTasks(1);
			     builder
			          .setBolt("bolt-splitter", new SocketBolt(), 1)
			          .shuffleGrouping("spout-producer")
			          .setNumTasks(1);
			     builder.setBolt("bolt-counter", new SocketCountBolt(), 3)
			          .fieldsGrouping("bolt-splitter", new Fields("productID"))
			          .setNumTasks(1);
			     // submit topology
			     Config conf = new Config();
			     String name = StartTopology.class.getSimpleName();
			     if (args != null && args.length > 0) {
			          String nimbus = args[0];
			          conf.put(Config.NIMBUS_HOST, nimbus);
			          conf.setNumWorkers(2);
			          StormSubmitter.submitTopologyWithProgressBar(name, conf, builder.createTopology());
			     } else {
			    	 //本地模式
			          LocalCluster cluster = new LocalCluster();
			          cluster.submitTopology(name, conf, builder.createTopology());
			  /*        Thread.sleep(60000);
			          cluster.shutdown();*/
			          //方法二
			          /*conf.setMaxTaskParallelism(1);
			          LocalCluster cluster = new LocalCluster();
			          cluster.submitTopology("simple", config, topologyBuilder.createTopology());*/
			     }

	}
}
